package org.example.petitionplatformsystem.service.Impl;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;
import org.example.petitionplatformsystem.dao.AuthenticationRepository;
import org.example.petitionplatformsystem.dao.PetitionEventsRepository;
import org.example.petitionplatformsystem.dao.UsersRepository;
import org.example.petitionplatformsystem.dao.model.Authentication;
import org.example.petitionplatformsystem.dao.model.PetitionEvents;
import org.example.petitionplatformsystem.dao.model.Users;
import org.example.petitionplatformsystem.service.AuthPetitionService;
import org.example.petitionplatformsystem.service.PetitionEventsService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Service
public class AuthPetitionServiceImpl implements AuthPetitionService {
    @Autowired
    private AuthenticationRepository authenticationRepository;
    @Autowired
    private UsersRepository usersRepository;
    @Autowired
    private PetitionEventsRepository petitionEventsRepository;
    @Autowired
    private SparkSession spark;
    @Override
    public Map<String, Object> getPetitionCountsByAddress() {
        // 从JPA读取数据
        List<PetitionEvents> events1 = petitionEventsRepository.findAll();
        Dataset<Row> df1 = spark.createDataFrame(events1, PetitionEvents.class);
        List<Users> events2 = usersRepository.findAll();
        Dataset<Row> df2 = spark.createDataFrame(events2, Users.class);
        List<Authentication> events3 = authenticationRepository.findAll();
        Dataset<Row> df3 = spark.createDataFrame(events3, Authentication.class);

        // 进行数据关联
        Dataset<Row> df = df1.join(df2, df1.col("UserID").equalTo(df2.col("UserID")))
                .join(df3, df1.col("UserID").equalTo(df3.col("UserID")));

        // 添加一个新的日期列，只包含年月日
        df = df.withColumn("DateOnly", functions.date_format(df.col("CreatedAt"), "yyyy-MM-dd"));

        // 统计不同Address的信访总数
        Dataset<Row> addressCount = df.groupBy("Address").count();
        Map<String, Long> addressPetitionCount = addressCount.collectAsList().stream()
                .collect(Collectors.toMap(row -> row.getString(0), row -> row.getLong(1)));

        // 统计不同Address的每天的信访量
        Dataset<Row> dailyCount = df.groupBy("Address", "DateOnly").count();
        Map<String, Map<String, Long>> dailyPetitionCount = new HashMap<>();
        dailyCount.collectAsList().forEach(row -> {
            String address = row.getString(0);
            String date = row.getString(1);
            long count = row.getLong(2);
            dailyPetitionCount.computeIfAbsent(address, k -> new HashMap<>()).put(date, count);
        });

        // 构建结果
        Map<String, Object> result = new HashMap<>();
        result.put("addressPetitionCount", addressPetitionCount);
        result.put("dailyPetitionCount", dailyPetitionCount);

        return result;
    }



}
